

package com.coding4play.mqtt.broker.protocol;

import com.coding4play.mqtt.broker.config.BrokerProperties;
import com.coding4play.mqtt.common.auth.IAuthService;
import com.coding4play.mqtt.common.entity.message.DupPubRelMessage;
import com.coding4play.mqtt.common.entity.message.DupPublishMessage;
import com.coding4play.mqtt.common.entity.session.ChannelSession;
import com.coding4play.mqtt.common.repository.IDupPubRelMessageRepository;
import com.coding4play.mqtt.common.repository.IDupPublishMessageRepository;
import com.coding4play.mqtt.common.repository.ISessionRepository;
import com.coding4play.mqtt.common.repository.ISubMessageRepository;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.*;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.AttributeKey;
import io.netty.util.CharsetUtil;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.StringUtils;

import java.util.List;

/**
 * CONNECT连接处理
 */
@AllArgsConstructor
@Slf4j
public class Connect {

    private ISessionRepository sessionRepository;
    private ISubMessageRepository subMessageRepository;
    private IDupPublishMessageRepository dupPublishMessageRepository;
    private IDupPubRelMessageRepository dupPubRelMessageRepository;
    private IAuthService authService;
    private BrokerProperties brokerProperties;

    public void processConnect(Channel channel, MqttConnectMessage msg) {
        String clientId = msg.payload().clientIdentifier();

        // clientId为空或null的情况, 这里要求客户端必须提供clientId, 不管cleanSession是否为1, 此处没有参考标准协议实现
        if (!StringUtils.hasLength(clientId)) {
            MqttConnAckMessage connAckMessage = (MqttConnAckMessage) MqttMessageFactory.newMessage(
                    new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
                    new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED, false), null);
            channel.writeAndFlush(connAckMessage);
            channel.close();
            return;
        }
        // 用户名和密码验证, 这里要求客户端连接时必须提供用户名和密码, 不管是否设置用户名标志和密码标志为1, 此处没有参考标准协议实现
        if (brokerProperties.getMqttPasswordMust()) {
            String username = msg.payload().userName();
            byte[] pwdBytes = msg.payload().passwordInBytes();
            String password = pwdBytes == null ? null : new String(pwdBytes, CharsetUtil.UTF_8);
            if (!authService.checkValid(username, password)) {
                MqttConnAckMessage connAckMessage = (MqttConnAckMessage) MqttMessageFactory.newMessage(
                        new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
                        new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD, false), null);
                channel.writeAndFlush(connAckMessage);
                channel.close();
                return;
            }
        }
        // 如果会话中已存储这个新连接的clientId, 就关闭之前该clientId的连接
        if (sessionRepository.hasSession(clientId)) {
            ChannelSession channelSession = sessionRepository.get(clientId);
            boolean cleanSession = channelSession.getCleanSession();
            if (cleanSession) {
                sessionRepository.delete(channelSession);
                subMessageRepository.delete(clientId);
                dupPublishMessageRepository.delete(clientId);
                dupPubRelMessageRepository.delete(clientId);
            }
        } else {
            //如果不存在session，则清除之前的其他缓存
            subMessageRepository.delete(clientId);
            dupPublishMessageRepository.delete(clientId);
            dupPubRelMessageRepository.delete(clientId);
        }

        // 处理连接心跳包
        int expire = 0;
        int keepAlive = msg.variableHeader().keepAliveTimeSeconds();
        if (keepAlive > 0) {
            if (channel.pipeline().names().contains("idle")) {
                channel.pipeline().remove("idle");
            }
            expire = Math.round(keepAlive * 1.5f);
            channel.pipeline().addFirst("idle", new IdleStateHandler(0, 0, expire));
        }

        // 处理遗嘱信息
        ChannelSession channelSession = ChannelSession.builder()
                .brokerId(brokerProperties.getId())
                .clientId(clientId)
                .channelId(channel.id().asLongText())
                .cleanSession(msg.variableHeader().isCleanSession())
                .willMessage(null)
                .expire(expire)
                .build();
        if (msg.variableHeader().isWillFlag()) {
            MqttPublishMessage willMessage = (MqttPublishMessage) MqttMessageFactory.newMessage(
                    new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.valueOf(msg.variableHeader().willQos()), msg.variableHeader().isWillRetain(), 0),
                    new MqttPublishVariableHeader(msg.payload().willTopic(), 0), Unpooled.buffer().writeBytes(msg.payload().willMessageInBytes()));
            channelSession.setWillMessage(willMessage);
        }

        // 至此存储会话信息及返回接受客户端连接
        sessionRepository.put(clientId, channelSession, expire);

        // 将clientId存储到channel的map中
        channel.attr(AttributeKey.valueOf("clientId")).set(clientId);
        boolean sessionPresent = sessionRepository.hasSession(clientId) && !msg.variableHeader().isCleanSession();
        MqttConnAckMessage okResp = (MqttConnAckMessage) MqttMessageFactory.newMessage(
                new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
                new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_ACCEPTED, sessionPresent), null);
        channel.writeAndFlush(okResp);
        log.debug("CONNECT - clientId: {}, cleanSession: {}", clientId, msg.variableHeader().isCleanSession());

        // 如果cleanSession为0, 需要重发同一clientId存储的未完成的QoS1和QoS2的DUP消息
        if (!msg.variableHeader().isCleanSession()) {
            List<DupPublishMessage> dupPublishMessageList = dupPublishMessageRepository.get(clientId);
            List<DupPubRelMessage> dupPubRelMessageList = dupPubRelMessageRepository.get(clientId);
            dupPublishMessageList.forEach(dupPublishMessageStore -> {
                MqttPublishMessage publishMessage = (MqttPublishMessage) MqttMessageFactory.newMessage(
                        new MqttFixedHeader(MqttMessageType.PUBLISH, true, MqttQoS.valueOf(dupPublishMessageStore.getMqttQoS()), false, 0),
                        new MqttPublishVariableHeader(dupPublishMessageStore.getTopic(), dupPublishMessageStore.getMessageId()), Unpooled.buffer().writeBytes(dupPublishMessageStore.getMessageBytes()));
                channel.writeAndFlush(publishMessage);
            });
            dupPubRelMessageList.forEach(dupPubRelMessageStore -> {
                MqttMessage pubRelMessage = MqttMessageFactory.newMessage(
                        new MqttFixedHeader(MqttMessageType.PUBREL, true, MqttQoS.AT_MOST_ONCE, false, 0),
                        MqttMessageIdVariableHeader.from(dupPubRelMessageStore.getMessageId()), null);
                channel.writeAndFlush(pubRelMessage);
            });
        }
    }

}
